/**
 * file: JobQueueServer.java
 */

package pcsd.server;

import java.rmi.RemoteException;
import java.util.ArrayDeque;
import java.util.Deque;

import pcsd.worker.JobProcessingWorker;
import pcsd.server.Monitor;
import pcsd.Task;

/**
 * A simple job queue server interface.
 * 
 * The server requires the clients to register themselves.
 * 
 * @author Tim van Deurzen, Benjamin Ertl, Justinas Murzinas, Boyan Mihaylov
 */
public class SimpleJobQueueServer implements JobQueueServer {
	
	/**
	 * Error codes for the interface.
	 */
	public static final int WORKER_THREADS_NOT_SPAWNED = -43;
	public static final int WORKER_THREADS_ALREADY_SPAWNED = -42;

	/** The number of workers. */
	private int numWorkers;

	/** The default maximum size of each queue per worker. */
	private int maxDequeSize;

	/** The task queues for each worker. */
	private Deque<Task>[] workerDeques;

	/** One thread per worker to dispatch tasks. */
	private Runnable[] taskDispatchers;
	
	/** Points to the worker which will get (eventually) the next task. */
	private int currentWorkerNum;

	/**
	 * Instantiates the service with a given number of workers.
	 */
	@SuppressWarnings("unchecked")
	public SimpleJobQueueServer(int numWorkers, int maxDequeSize) {
		this.numWorkers = numWorkers;
		this.maxDequeSize = maxDequeSize;
		this.workerDeques = new Deque[numWorkers];
		this.currentWorkerNum = 0;

		for (int i = 0; i < numWorkers; i++) {
			workerDeques[i] = new ArrayDeque<Task>();
		}
	}

	/** 
	 * Enqueue a new task adding it to one of the workers' queue in a round
	 * robin fashion.
	 *
	 * This function blocks until it finds a spot for the task. This may lead
	 * to a time-out of the RMI sub-system.
	 */
	@Override
	public final int enqueueTask(Task task) throws RemoteException {
	    if (this.isWorkersSpawned()) {
		    // Enqueue the task safely.
		    this.enqueueTaskSafe(task);
	    	return 0;
	    } 
	    else {
	    	// The workers are not spawned yet.
	    	return WORKER_THREADS_NOT_SPAWNED;
	    }
	}

    /**
     * Start all the task dispatchers for the workers.
     */
	public final int spawnWorkerThreads(JobProcessingWorker[] workers, Monitor monitor)
			throws RemoteException {
	    if (this.isWorkersSpawned()) {
	    	// The workers are already spawned
	    	return WORKER_THREADS_ALREADY_SPAWNED;
	    } 
	    else {
	    	// Create dispatcher for each worker and start it in a new thread
	    	TaskDispatcher[] dispatchers = new TaskDispatcher[workers.length];
	    	for (int i = 0; i < workers.length; i++) {
	    		dispatchers[i] = new TaskDispatcher(this.workerDeques[i],
	    				workers[i], i, monitor);
	    		
	    		new Thread(dispatchers[i]).start();
	    	}
	    	
	    	// Initialize task dispatchers
	    	this.taskDispatchers = dispatchers;
	    	return 0;
	    }
	}
	
	/**
	 * Enqueues a task.
	 * @param task
	 * @throws RemoteException
	 */
	protected void enqueueTaskSafe(Task task) throws RemoteException {
		Boolean added = false;
		
		while (true) {
			synchronized(this) {
				// Find the first worker after currentWorkerNum which
				// has a room for the new task.
		    	int startWorkerNum = this.currentWorkerNum;
		    	
		    	do {
		    		// Here we do a double check. We rely on the fact that
		    		// in most cases other threads will only remove tasks.
		    		// Only the case when we have multiple clients may result
		    		// in multiple threads adding new tasks. So, why acquire a lock
		    		// when it may not be needed.
		    		if (this.workerDeques[currentWorkerNum].size() < maxDequeSize) {
		    			synchronized (this.workerDeques[this.currentWorkerNum]) {
			    			// Check whether there is a room for the task in this queue
				    		if (this.workerDeques[this.currentWorkerNum].size() < maxDequeSize) {
				    			this.workerDeques[this.currentWorkerNum].push(task);
				    			added = true;
				    		}
			    		}
		    		}
		    		
		    		// Move the pointer forward
		    		this.currentWorkerNum++;
		    		if (this.currentWorkerNum == this.numWorkers) {
		    			this.currentWorkerNum = 0;
		    		}
		    		
		    		// The task is successfully added.
		    		if (added) break;
		    	}
		    	while (this.currentWorkerNum != startWorkerNum);
			}
    	
			if (added) {
				// The task is successfully added.
				break;
			}
			else {
				// All queues are full so we go to sleep for a while
	    		System.out.println("Queue limit reached. Waiting...");
	    		try {
	    			Thread.sleep(1000L);
	    		}
	    		catch (InterruptedException ie) {
	    			throw new RemoteException("EnqueueTask died :(");
	    		}
	    	}
		}
	}
	
	/**
	 * Gets all worker queues.
	 * @return
	 */
	protected final Deque<Task>[] getWorkerQueues() {
		return this.workerDeques;
	}
	
	/**
	 * Checks whether the task dispatchers have been registered.
	 * @return
	 */
	private Boolean isWorkersSpawned() {
		return this.taskDispatchers != null;
	}

	
}
